package com.hzmg.handler.http;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.http.javadsl.Http;
import akka.http.javadsl.model.*;
import akka.stream.javadsl.Source;
import akka.util.ByteString;
import com.hzmg.handler.factory.RequestDealFactory;
import com.hzmg.route.AkkaResponseEntity;
import com.hzmg.route.Shift;
import com.hzmg.route.RouteDefinition;
import com.hzmg.support.util.NameUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;


public class AkkaHttpRoutingHandler extends AbstractBehavior<Shift> {
    private static final Logger logger = LoggerFactory.getLogger(AkkaHttpRoutingHandler.class);

    private final Http http = Http.get(getContext().getSystem());

    public AkkaHttpRoutingHandler(ActorContext<Shift> context) {
        super(context);
    }

    public static Behavior<Shift> create() {
        return Behaviors.setup(context -> new AkkaHttpRoutingHandler(context));
    }

    @Override
    public Receive<Shift> createReceive() {
        return newReceiveBuilder()
                .onMessage(RouteDefinition.class, this::onRequest)
                .build();
    }

    private Behavior<Shift> onRequest(RouteDefinition routeDefinition) throws ExecutionException, InterruptedException {
        CompletableFuture<HttpResponse> httpResponseFuture = fetch(routeDefinition).toCompletableFuture();
        HttpResponse response = httpResponseFuture.get();
        ActorRef<Shift> sendResponse = getContext().spawn(AkkaHttpWriteResponseHandler.create(),
                NameUtils.normalizeActorRefSimpleName(AkkaHttpWriteResponseHandler.class));

        logger.info("[akkacloud-gateway-core] AkkaHttpRoutingHandler status: {}"
                , response.status());
        AkkaResponseEntity responseEntity = new AkkaResponseEntity(response);
        sendResponse.tell(responseEntity);
        return Behaviors.same();
    }

    CompletionStage<HttpResponse> fetch(RouteDefinition routeDefinition) {
        logger.info("[akkacloud-gateway-core] AkkaHttpRoutingHandler routeDefinition: {}", routeDefinition);
        HttpRequest request = routeDefinition.getRequest();

        HttpMethod method = request.method();
        //TODO: 请求URI处理
        //TODO: 请求参数定义
        logger.info("[akkacloud-gateway-core] AkkaHttpRoutingHandler method: {}", method.name());
        RequestDeal requestDeal = RequestDealFactory.getDealViaRequestMethod(method.name());
        HttpRequest sendRequest = requestDeal.viaRequestMethod(request);

        logger.info("[akkacloud-gateway-core] AkkaHttpRoutingHandler E: {}", sendRequest.entity());
        return http.singleRequest(sendRequest);
    }

    private Source<ByteString, ?> extracEntityData(HttpResponse httpResponse) {
        if (httpResponse.status() == StatusCodes.OK) {
            return httpResponse.entity().getDataBytes();
        } else {
            return Source.failed(new RuntimeException("illegal response " + httpResponse));
        }
    }

}
